﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Web;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using System.Threading;
using System.Net.Http.Formatting;
using System.Net.Http;

namespace OrderProcessing
{
    public sealed class ShipProcessor<T> : IPropagatorBlock<T, T>, ITargetBlock<T>, IReceivableSourceBlock<T>, ISourceBlock<T>, IDataflowBlock
    {
        private Timer _scheduler;
        private IReceivableSourceBlock<T> _source;
        private ITargetBlock<T> _target;
        public ShipProcessor(int interval)
        {
            var source = new BufferBlock<T>();
            var pendingBlock=new BufferBlock<T>();
            var subscriber = new ActionBlock<T>(async (item) =>
                {
                    var client = new System.Net.Http.HttpClient();
                    var response = await client.PostAsync("http://arch-lpt-win-02.qa.nt.ctripcorp.com:9099/api/ship", item, new JsonMediaTypeFormatter());
                    if (!response.IsSuccessStatusCode)
                        throw new ApplicationException("Exception in shipping processor (" + item + "): " + response.ReasonPhrase);

                    bool result = await response.Content.ReadAsAsync<bool>();
                    if(!result)
                    {
                        pendingBlock.Post(item);
                    } }
                );
            source.LinkTo(subscriber);
            _scheduler = new Timer(new TimerCallback((state) =>
            {
                var item = pendingBlock.Receive();
                subscriber.Post(item);
                _scheduler.Change(interval, Timeout.Infinite);

            }), null, interval, Timeout.Infinite);
        }
        public DataflowMessageStatus OfferMessage(DataflowMessageHeader messageHeader, T messageValue, ISourceBlock<T> source, bool consumeToAccept)
        {
            throw new NotImplementedException();
        }

        public void Complete()
        {
            throw new NotImplementedException();
        }

        public Task Completion
        {
            get { throw new NotImplementedException(); }
        }

        public void Fault(Exception exception)
        {
            throw new NotImplementedException();
        }

        public T ConsumeMessage(DataflowMessageHeader messageHeader, ITargetBlock<T> target, out bool messageConsumed)
        {
            throw new NotImplementedException();
        }

        public IDisposable LinkTo(ITargetBlock<T> target, DataflowLinkOptions linkOptions)
        {
            throw new NotImplementedException();
        }

        public void ReleaseReservation(DataflowMessageHeader messageHeader, ITargetBlock<T> target)
        {
            throw new NotImplementedException();
        }

        public bool ReserveMessage(DataflowMessageHeader messageHeader, ITargetBlock<T> target)
        {
            throw new NotImplementedException();
        }

        public bool TryReceive(Predicate<T> filter, out T item)
        {
            throw new NotImplementedException();
        }

        public bool TryReceiveAll(out IList<T> items)
        {
            throw new NotImplementedException();
        }
    }
}